package com.edee.bigday.seckill.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.edee.bigday.seckill.feign.CommodityFeignService;
import com.edee.bigday.seckill.feign.CouponFeignService;
import com.edee.bigday.seckill.interceptor.LoginUserInterceptor;
import com.edee.bigday.seckill.service.SecKillService;
import com.edee.bigday.seckill.to.SeckillSkuRedisTo;
import com.edee.bigday.seckill.vo.SeckillSessionsWithSkus;
import com.edee.bigday.seckill.vo.SeckillSkuVo;
import com.edee.bigday.seckill.vo.SkuInfoVo;
import com.edee.common.to.SeckillOrderTo;
import com.edee.common.utils.R;
import com.edee.common.vo.MemberRespVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Slf4j
@Service
public class SecKillServiceImpl implements SecKillService {

    @Autowired
    CouponFeignService couponFeignService;
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    CommodityFeignService commodityFeignService;
    @Autowired
    RedissonClient redissonClient;
    @Autowired
    RabbitTemplate rabbitTemplate;

    private final String SESSIONS_CACHE_PREFIX = "seckill:sessions:";
    private final String SECKILL_SKU_CACHE_PREFIX = "seckill:skus";
    private final String SKU_STOCK_SEMAPHORE = "seckill:stock:";

    @Override
    public void uploadLatestThreeDaysSeckillSku() {
        //1、扫描数据表中最近三天需要参与秒杀的活动，需要远程调用优惠系统
        R latestThreeDaysSession = couponFeignService.getLatestThreeDaysSession();
        if (latestThreeDaysSession.getCode() == 0) {
            //上架商品
            //1、获取上架商品信息(问题：Data中的数据转换后，时间增加了8小时)
            List<SeckillSessionsWithSkus> sessionDataList = latestThreeDaysSession.getData(new TypeReference<>() {
            });
            //2、将上架商品信息缓存到Redis中
            //2.1、缓存秒杀活动信息
            saveSessionInfos(sessionDataList);
            //2.2、缓存的关联商品信息
            saveSessionSkuInfos(sessionDataList);
        }
    }

    /**
     * 获取当前时间可以参与秒杀的商品信息
     *
     * @return
     */
    @Override
    public List<SeckillSkuRedisTo> getCurrentSeckillSkus() {
        //1、查询当前时间内存在哪些秒杀场次
        long now = new Date().getTime();//当前时间与1970-01-01之间的差
        Set<String> keys = stringRedisTemplate.keys(SESSIONS_CACHE_PREFIX + "*");
        for (String key : keys) {
            String replace = key.replace(SESSIONS_CACHE_PREFIX, "");
            String[] split = replace.split("_");
            long startTime = Long.parseLong(split[0]);
            long endTime = Long.parseLong(split[1]);
            if (now >= startTime && now <= endTime) {
                //由于CentOS系统时间不准，为了测试效果，暂时修改如下（2021.10.31已修正，修正失败，后强制减去8小时）：
                //从数据库中取出的数据在上传到Redis时，加了8h：
                //if (now <= endTime) {
                //2、获取当前时间段内的秒杀场次所需要的所有商品信息
                //（range()方法的使用可以通过方法中提供的https://redis.io/commands/lrange查阅）
                List<String> range = stringRedisTemplate.opsForList().range(key, -100, 100);
                BoundHashOperations<String, String, String> hashOps = stringRedisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
                List<String> list = hashOps.multiGet(range);
                //上面转换数据之后，时间增加8小时
                if (CollectionUtils.isNotEmpty(list)) {
                    List<SeckillSkuRedisTo> seckillSkuRedisToList = list.stream().map(item -> {
                        SeckillSkuRedisTo skuRedisTo = JSON.parseObject(item, SeckillSkuRedisTo.class);
                        //skuRedisTo.setRandomCode(null);//秒杀开始前，要讲随机码屏蔽掉
                        skuRedisTo.setStartTime(skuRedisTo.getStartTime() - 8 * 60 * 60 * 1000);
                        skuRedisTo.setEndTime(skuRedisTo.getEndTime() - 8 * 60 * 60 * 1000);
                        return skuRedisTo;
                    }).collect(Collectors.toList());
                    return seckillSkuRedisToList;
                }
                //获取到当前秒杀场次后，其余的无须继续遍历
                break;
            }
        }
        return null;
    }

    @Override
    public SeckillSkuRedisTo getSkuSeckillInfo(Long skuId) {
        // 在Redis中查询所有需要参与秒杀的商品的key
        BoundHashOperations<String, String, String> hashOps = stringRedisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
        Set<String> keys = hashOps.keys();
        if (CollectionUtils.isNotEmpty(keys)) {
            //Pattern.matches("\\d*[0-9]_33", "02134234_33")
            //原来是String regx = "\\d_" + skuId; 无法匹配多位数，如：只能匹配5_33，不能匹配10_33
            String regx = "\\d*[0-9]_" + skuId;
            for (String key : keys) {
                if (Pattern.matches(regx, key)) {
                    String json = hashOps.get(key);
                    SeckillSkuRedisTo seckillSkuRedisTo = JSON.parseObject(json, SeckillSkuRedisTo.class);
                    //根据秒杀信息判断当前时间段是否处于秒杀时间内，处理随机码
                    Long startTime = seckillSkuRedisTo.getStartTime();
                    Long endTime = seckillSkuRedisTo.getEndTime();
                    long now = new Date().getTime();
                    if (now < startTime && now > endTime) {
                        seckillSkuRedisTo.setRandomCode(null);
                    }
                    return seckillSkuRedisTo;
                }
            }
        }
        return null;
    }

    /**
     * 秒杀操作
     *
     * @param killId
     * @param key
     * @param num
     * @return
     */
    @Override
    public String kill(String killId, String key, Integer num) {
        // 执行开始时间
        long timeBegin = System.currentTimeMillis();
        //1、登录判断，由拦截器LoginUserInterceptor来负责。
        MemberRespVo memberRespVo = LoginUserInterceptor.loginUser.get();
        //2、合法性校验
        //获取当前秒杀商品的详细信息
        BoundHashOperations<String, String, String> hashOps = stringRedisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
        String str = hashOps.get(killId);
        if (StringUtils.isNotEmpty(str)) {
            //将json数据转换为指定的类型
            SeckillSkuRedisTo seckillSkuRedisTo = JSON.parseObject(str, SeckillSkuRedisTo.class);
            //校验合法性
            Long startTime = seckillSkuRedisTo.getStartTime() - 8 * 60 * 60 * 1000;
            Long endTime = seckillSkuRedisTo.getEndTime() - 8 * 60 * 60 * 1000;
            long now = new Date().getTime();
            //校验时间的合法性
            if (now >= startTime && now <= endTime) {
                //校验商品Id和随机码
                String killIdFromRedis = seckillSkuRedisTo.getPromotionSessionId() + "_" + seckillSkuRedisTo.getSkuId();
                String randomCode = seckillSkuRedisTo.getRandomCode();
                if (killIdFromRedis.equals(killId) && randomCode.equals(key)) {
                    //校验秒杀数量是否合理
                    Integer seckillLimit = seckillSkuRedisTo.getSeckillLimit();
                    if (num <= seckillLimit) {
                        //校验当前用户是否已经购买过该商品（幂等性，只要秒杀成功，就用一个唯一key去Redis中占位，如suerId_sessionId_skuId）
                        String redisPlaceHolder = memberRespVo.getId() + "_" + killIdFromRedis;
                        long ttl = endTime - now;
                        Boolean setIfAbsent = stringRedisTemplate.opsForValue().setIfAbsent(redisPlaceHolder, num.toString(), ttl, TimeUnit.MILLISECONDS);
                        if (setIfAbsent) {
                            //占位成功，说明当前用户未曾购买过该商品
                            //使用分布式信号量扣减库存
                            RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + randomCode);
                            try {
                                boolean tryAcquire = semaphore.tryAcquire(num, 100, TimeUnit.MILLISECONDS);
                                //boolean tryAcquire = semaphore.tryAcquire(num);使用tryAcquire尝试获取信号量，就无须再使用try catch。
                                if (tryAcquire) {
                                    //秒杀成功
                                    //快速下单：发送MQ消息
                                    //生成并返回订单号（使用IdWorker生成订单号）
                                    String orderSn = IdWorker.getTimeId();
                                    //设置秒杀订单信息
                                    SeckillOrderTo seckillOrderTo = new SeckillOrderTo();
                                    seckillOrderTo.setOrderSn(orderSn);
                                    seckillOrderTo.setSkuId(seckillSkuRedisTo.getSkuId());
                                    seckillOrderTo.setNum(num);
                                    seckillOrderTo.setPromotionSessionId(seckillSkuRedisTo.getPromotionSessionId());
                                    seckillOrderTo.setMemberId(memberRespVo.getId());
                                    seckillOrderTo.setSeckillPrice(seckillSkuRedisTo.getSeckillPrice());
                                    rabbitTemplate.convertAndSend("order-event-exchange", "order.seckill.order", seckillOrderTo);
                                    // 执行结束时间
                                    long timeEnd = System.currentTimeMillis();
                                    log.info("方法耗时：{}ms。",timeEnd-timeBegin);
                                    return orderSn;
                                }
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                                return null;
                            }
                        }
                    }
                }
            }
        }
        return null;
    }

    private void saveSessionInfos(List<SeckillSessionsWithSkus> sessionDataList) {
        if (CollectionUtils.isNotEmpty(sessionDataList)){
            sessionDataList.stream().forEach(sessionData -> {
                //获取秒杀起止时间，并减去转换数据过程中增加的8小时
                Long startTime = sessionData.getStartTime().getTime() - 8 * 60 * 60 * 1000;
                Long endTime = sessionData.getEndTime().getTime() - 8 * 60 * 60 * 1000;
                //拼接key
                String key = SESSIONS_CACHE_PREFIX + startTime + "_" + endTime;
                // 幂等性保证
                Boolean hasKey = stringRedisTemplate.hasKey(key);
                if (!hasKey) {
                    List<String> seckillSkuIds = sessionData.getSkuRelationEntities().stream().map(item ->
                            item.getPromotionSessionId() + "_" + item.getSkuId()).collect(Collectors.toList());
                    //缓存活动信息
                    if (CollectionUtils.isNotEmpty(seckillSkuIds)) {
                        stringRedisTemplate.opsForList().leftPushAll(key, seckillSkuIds);
                    }
                }
            });
        }
    }

    private void saveSessionSkuInfos(List<SeckillSessionsWithSkus> sessionDataList) {
        sessionDataList.stream().forEach(sessionData -> {
            //准备Hash操作的结构
            BoundHashOperations<String, Object, Object> boundHashOps = stringRedisTemplate.boundHashOps(SECKILL_SKU_CACHE_PREFIX);
            List<SeckillSkuVo> skuRelationEntities = sessionData.getSkuRelationEntities();
            skuRelationEntities.stream().forEach(seckillSkuVo -> {
                // 生成随机码token
                String token = UUID.randomUUID().toString().replace("-", "");
                // 保证商品信息的、秒杀场次时间信息、库存信号量的幂等性
                if (!boundHashOps.hasKey(seckillSkuVo.getPromotionSessionId() + "_" + seckillSkuVo.getSkuId())) {
                    //缓存商品信息
                    SeckillSkuRedisTo redisTo = new SeckillSkuRedisTo();
                    //1、缓存商品基本信息
                    R skuInfo = commodityFeignService.getSkuInfo(seckillSkuVo.getSkuId());
                    if (skuInfo.getCode() == 0) {
                        SkuInfoVo info = skuInfo.getData("skuInfo", new TypeReference<>() {
                        });
                        redisTo.setSkuInfo(info);
                    }
                    //2、缓存秒杀信息
                    BeanUtils.copyProperties(seckillSkuVo, redisTo);

                    //3、缓存秒杀商品的时间信息
                    long startTime = sessionData.getStartTime().getTime();
                    long endTime = sessionData.getEndTime().getTime();
                    redisTo.setStartTime(startTime);
                    redisTo.setEndTime(endTime);
                    //4、设置商品的随机码（防止工具自动访问、内部人员作弊等攻击）
                    redisTo.setRandomCode(token);
                    String redisToToString = JSON.toJSONString(redisTo);
                    boundHashOps.put(seckillSkuVo.getPromotionSessionId() + "_" + seckillSkuVo.getSkuId(), redisToToString);
                    //5、设置秒杀商品分布式信号量，作为库存扣减信息（主要作用之一：限流）
                    //获取信号量
                    RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + token);
                    //设置信号量（可以秒杀的商品数量作为信号量）
                    Integer seckillCount = seckillSkuVo.getSeckillCount();
                    semaphore.trySetPermits(seckillCount);
                }
            });
        });
    }
}
